/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *    Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.transfer;

import com.dtdream.emr.transmission.data.Record;
import com.dtdream.emr.transmission.exception.TransmissionException;
import com.dtdream.emr.transmission.io.Constant;
import com.dtdream.emr.transmission.io.Request;
import io.netty.channel.EventLoopGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_NETWORK;
import static com.dtdream.emr.transmission.io.Constant.URL_INIT;

/**
 * Created by moriarty on 17-6-12.
 */
public class TransferNode implements AutoCloseable {
    private static final Logger LOG = LoggerFactory.getLogger(TransferNode.class);

    private InetSocketAddress address;
    private ControlChannel controlChannel;
    private DataChannel<Request> dataChannel;
    private String accessKey;
    private String keySecret;
    private EventLoopGroup workgroup;
    private NettyConfig config;

    public TransferNode(InetSocketAddress socketAddress,
                        String accessKey,
                        String keySecret,
                        EventLoopGroup workgroup,
                        NettyConfig nettyConfig) throws TransmissionException {
        this.address = socketAddress;
        this.controlChannel = new ControlChannel(socketAddress.getHostString(), socketAddress.getPort());
        this.accessKey = accessKey;
        this.keySecret = keySecret;
        this.workgroup = workgroup;
        this.config = nettyConfig;
    }

    public void init() throws TransmissionException {
        try {
            Map<String, Object> response = this.post(URL_INIT, null);
            int dataPort = Integer.parseInt(response.get(Constant.DATA_PORT).toString());

            this.dataChannel = new DataChannel<>(address.getHostString(), dataPort, workgroup, config);
            dataChannel.init();
        } catch (IOException e) {
            throw new TransmissionException(EXCEPTION_SDK_NETWORK, e);
        }
    }

    public InetSocketAddress address() {
        return address;
    }

    List<String> controlServers() throws IOException, TransmissionException {
        Map<String, Object> params = new HashMap<>();
        params.put(Constant.ACCESSKEY, accessKey);
        params.put(Constant.KEYSECRET, keySecret);
        return (List<String>) controlChannel
                .get(Constant.URL_GET_CONTROL_SERVERS, params)
                .get(Constant.CONTROL_SERVERS);
    }

    private static int TRY_TIMES = 3;
    private static int RETRY_INTERVAL_MILLS = 3000;

    public Map<String, Object> post(String path, Map<String, Object> params) throws TransmissionException, IOException {
        if (null == params) {
            params = new HashMap<>();
        }
        params.put(Constant.ACCESSKEY, accessKey);
        params.put(Constant.KEYSECRET, keySecret);

        int tryTimes = 0;
        while (true) {
            try {
                return controlChannel.post(path, params);
            } catch (IOException e) {
                if ((e instanceof SSLException) || (++tryTimes >= TRY_TIMES)) {
                    throw e;
                }

                try {
                    LOG.warn("Post error, retry later", e);
                    Thread.sleep(RETRY_INTERVAL_MILLS);
                } catch (InterruptedException e1) {
                    LOG.warn("Waiting was interrupted", e);
                }
            }
        }
    }

    public void writeAsync(UUID jobId, int id, List<Record> recordList) throws TransmissionException {
        Request request = new Request();
        request.setRecord(recordList);
        request.setJobId(jobId.toString());
        request.setId(id);
        request.setMessageType(Constant.WRITE_RECORD);

        dataChannel.send(request);
    }

    public void finishWriteAsync(UUID jobId, int id) throws TransmissionException {
        Request request = new Request();
        request.setJobId(jobId.toString());
        request.setId(id);
        request.setMessageType(Constant.WRITE_FINISHED);

        dataChannel.send(request);
    }

    public void readAsync(UUID jobId, int id) throws TransmissionException {
        Request request = new Request();
        request.setJobId(jobId.toString());
        request.setId(id);
        request.setMessageType(Constant.READ_RECORD);

        dataChannel.send(request);
    }

    public void close() {
        if (null != controlChannel) {
            controlChannel.close();
        }
        if (null != dataChannel) {
            dataChannel.close();
        }
    }
}